// Concurrent DVC operations tests
// Tests multiple users performing DVC operations simultaneously

use serde_json::Value;
use std::collections::HashMap;
use std::sync::Arc;
use std::time::Duration;
use tempfile::TempDir;
use tokio::task::JoinSet;
use tokio::time::timeout;

use axum::{
    extract::DefaultBodyLimit,
    http::Method,
    routing::{delete, get, post, put},
    Router,
};
use rdm_rust::config::AppConfig;
use rdm_rust::database::connection::establish_connection_pool;
use rdm_rust::handlers::{auth, data, dvc_remote};
use rdm_rust::services::{AuthService, DataService, DvcService, StorageService};
use rdm_rust::AppState;
use tower::ServiceBuilder;
use tower_http::{
    cors::{Any, CorsLayer},
    trace::TraceLayer,
};

// Test configuration
const TEST_SERVER_PORT: u16 = 0; // Use random available port
const TEST_TIMEOUT: Duration = Duration::from_secs(60); // Longer timeout for concurrent tests

struct ConcurrentDvcTestServer {
    port: u16,
    temp_dir: TempDir,
    base_url: String,
    client: reqwest::Client,
}

impl ConcurrentDvcTestServer {
    async fn new() -> Self {
        // Create temporary directory for test storage
        let temp_dir = TempDir::new().expect("Failed to create temp dir");

        // Load test configuration
        let mut config = AppConfig::from_env().expect("Failed to load config");
        config.server.port = TEST_SERVER_PORT;
        config.dvc.storage_path = temp_dir.path().to_string_lossy().to_string();

        // Initialize database
        let db_pool = establish_connection_pool(&config.database.url)
            .await
            .expect("Failed to establish database connection");

        // Initialize services
        let auth_service = Arc::new(AuthService::new(db_pool.clone(), config.clone()));
        let data_service = Arc::new(DataService::new(db_pool.clone()));
        let storage_service = Arc::new(StorageService::new(config.dvc.storage_path.clone()));
        let dvc_service = Arc::new(DvcService::new(db_pool.clone()));

        let app_state = AppState {
            auth_service,
            data_service,
            storage_service,
            dvc_service,
        };

        // Build application router
        let public_routes = Router::new()
            .route("/api/register", post(auth::register))
            .route("/api/register-admin", post(auth::register_admin))
            .route("/api/login", post(auth::login))
            .route("/api/data/public", get(data::list_public_data_items));

        let protected_routes = Router::new()
            .route("/api/user-info", get(auth::user_info))
            .route("/api/data", get(data::list_data_items))
            .route("/api/data/upload", post(data::upload_data))
            .route("/api/data/:id", get(data::get_data_item))
            .route("/api/data/:id", delete(data::delete_data_item))
            .route("/api/data/:id/download", get(data::download_data_file))
            .route("/api/data/:id/dvc", get(data::download_dvc_file))
            .route("/api/data/:id/dvc/content", get(data::get_dvc_content))
            .route("/api/data/:id/metadata", put(data::upload_metadata))
            .route("/api/dvc/files", get(dvc_remote::dvc_list_files))
            .route("/api/dvc/file/*path", get(dvc_remote::dvc_get_file_info))
            .route(
                "/api/dvc/remote/*path",
                axum::routing::any(dvc_remote::handle_dvc_request),
            )
            .route_layer(axum::middleware::from_fn_with_state(
                app_state.clone(),
                rdm_rust::middleware::auth::auth_middleware,
            ));

        let app = Router::new()
            .merge(public_routes)
            .merge(protected_routes)
            .layer(
                ServiceBuilder::new()
                    .layer(TraceLayer::new_for_http())
                    .layer(
                        CorsLayer::new()
                            .allow_origin(Any)
                            .allow_methods([
                                Method::GET,
                                Method::POST,
                                Method::PUT,
                                Method::DELETE,
                                Method::OPTIONS,
                                Method::HEAD,
                            ])
                            .allow_headers(Any),
                    )
                    .layer(DefaultBodyLimit::max(100 * 1024 * 1024)), // 100MB max file size
            )
            .with_state(app_state);

        // Start server on random port
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
            .await
            .expect("Failed to bind to address");
        let port = listener
            .local_addr()
            .expect("Failed to get local address")
            .port();

        // Spawn server task
        tokio::spawn(async move {
            axum::serve(listener, app)
                .await
                .expect("Failed to serve server");
        });

        // Give server time to start
        tokio::time::sleep(Duration::from_millis(100)).await;

        let base_url = format!("http://127.0.0.1:{}", port);
        let client = reqwest::Client::builder()
            .timeout(TEST_TIMEOUT)
            .build()
            .expect("Failed to create HTTP client");

        Self {
            port,
            temp_dir,
            base_url,
            client,
        }
    }

    async fn wait_for_server(&self) -> Result<(), Box<dyn std::error::Error>> {
        let health_url = format!("{}/api/data/public", self.base_url);

        for _ in 0..10 {
            match timeout(Duration::from_secs(1), self.client.get(&health_url).send()).await {
                Ok(Ok(response)) if response.status().is_success() => return Ok(()),
                Ok(_) => (),
                Err(_) => (),
            }
            tokio::time::sleep(Duration::from_millis(100)).await;
        }

        Err("Server failed to start".into())
    }

    async fn register_and_login(&self, username: &str, password: &str) -> String {
        // Register user
        let email = format!("{}@hillstonenet.com", username);
        let register_response = self
            .client
            .post(format!("{}/api/register", self.base_url))
            .json(&serde_json::json!({
                "email": email,
                "password": password
            }))
            .send()
            .await
            .expect("Failed to send register request");

        assert_eq!(
            register_response.status(),
            201,
            "User registration should succeed"
        );

        // Login
        let login_response = self
            .client
            .post(format!("{}/api/login", self.base_url))
            .json(&serde_json::json!({
                "email": email,
                "password": password
            }))
            .send()
            .await
            .expect("Failed to send login request");

        assert_eq!(login_response.status(), 200, "Login should succeed");

        let login_body: Value = login_response
            .json()
            .await
            .expect("Failed to parse login response");
        login_body
            .get("access_token")
            .or_else(|| login_body.get("token"))
            .and_then(|t| t.as_str())
            .expect("Login response should contain token")
            .to_string()
    }
}

#[tokio::test]
async fn test_concurrent_dvc_operations_multiple_users() {
    let server = ConcurrentDvcTestServer::new().await;
    server
        .wait_for_server()
        .await
        .expect("Test server failed to start");

    println!("Concurrent DVC test server running on: {}", server.base_url);

    // Create multiple users
    let users = vec![
        ("alice", "alice123"),
        ("bob", "bob123"),
        ("charlie", "charlie123"),
        ("diana", "diana123"),
        ("eve", "eve123"),
    ];

    // Register and login all users concurrently
    let mut join_set = JoinSet::new();
    let mut auth_tokens = HashMap::new();

    for (username, password) in &users {
        let server_clone = server.clone();
        let username = username.to_string();
        let password = password.to_string();

        join_set.spawn(async move {
            let token = server_clone.register_and_login(&username, &password).await;
            (username, token)
        });
    }

    while let Some(result) = join_set.join_next().await {
        let (username, token) = result.expect("Task should complete successfully");
        auth_tokens.insert(username, token);
    }

    println!(
        "Successfully registered and authenticated {} users",
        users.len()
    );

    // Each user pushes multiple files concurrently
    let mut push_tasks = JoinSet::new();

    for user_idx in 0..users.len() {
        let (username, _password) = users[user_idx].clone();
        let token = auth_tokens.get(&username[..]).unwrap().clone();
        let server_clone = server.clone();

        // Each user pushes 3 files
        for file_idx in 1..=3 {
            let server_clone = server_clone.clone();
            let token = token.clone();
            let username_clone = username.clone();

            push_tasks.spawn(async move {
                let file_path = format!("data/{}/file_{}.csv", username_clone, file_idx);
                let file_content = format!(
                    "Content for {} file {} from user {}",
                    file_path, file_idx, username_clone
                );
                let dvc_put_response = server_clone
                    .client
                    .put(format!(
                        "{}/api/dvc/remote/{}",
                        server_clone.base_url, file_path
                    ))
                    .header("Authorization", format!("Bearer {}", token))
                    .header("Content-Type", "application/octet-stream")
                    .body(file_content.into_bytes())
                    .send()
                    .await
                    .expect("Failed to send DVC PUT request");

                (username.clone(), file_path, dvc_put_response.status())
            });
        }
    }

    // Collect all push results
    let mut successful_pushes = 0;
    let total_pushes = users.len() * 3;

    while let Some(result) = push_tasks.join_next().await {
        let (username, file_path, status) = result.expect("Push task should complete successfully");
        assert_eq!(
            status, 200,
            "DVC PUT should succeed for {} from user {}",
            file_path, username
        );
        successful_pushes += 1;
    }

    assert_eq!(
        successful_pushes, total_pushes,
        "All DVC pushes should succeed"
    );
    println!(
        "Successfully completed {} concurrent DVC pushes",
        successful_pushes
    );

    // Verify each user can only see their own files
    for (username, _password) in &users {
        let token = auth_tokens.get(*username).unwrap();

        let dvc_list_response = server
            .client
            .get(format!("{}/api/dvc/files", server.base_url))
            .header("Authorization", format!("Bearer {}", token))
            .send()
            .await
            .expect("Failed to send DVC list request");

        assert_eq!(
            dvc_list_response.status(),
            200,
            "DVC list should succeed for user {}",
            username
        );

        let dvc_list_body: Value = dvc_list_response
            .json()
            .await
            .expect("Failed to parse DVC list response");
        let files = dvc_list_body
            .as_array()
            .expect("DVC list should return an array");

        // Each user should have exactly 3 files
        assert_eq!(
            files.len(),
            3,
            "User {} should have exactly 3 files",
            username
        );

        // All files should belong to this user
        for file in files {
            let file_path = file.get("path").unwrap().as_str().unwrap();
            assert!(
                file_path.contains(&format!("data/{}/", username)),
                "File {} should belong to user {}",
                file_path,
                username
            );
        }
    }

    println!("Verified user isolation - each user can only see their own files");
}

#[tokio::test]
async fn test_concurrent_dvc_operations_same_user() {
    let server = ConcurrentDvcTestServer::new().await;
    server
        .wait_for_server()
        .await
        .expect("Test server failed to start");

    // Register and login a single user
    let token = server
        .register_and_login("concurrent_user", "password123")
        .await;

    // Create multiple concurrent push operations for the same user
    let mut push_tasks = JoinSet::new();
    let file_count = 20; // Push 20 files concurrently

    for file_idx in 1..=file_count {
        let file_path = format!("data/concurrent/file_{}.dat", file_idx);
        let file_content = format!("Concurrent file {} content", file_idx);
        let server_clone = server.clone();
        let token = token.clone();

        push_tasks.spawn(async move {
            let dvc_put_response = server_clone
                .client
                .put(format!(
                    "{}/api/dvc/remote/{}",
                    server_clone.base_url, file_path
                ))
                .header("Authorization", format!("Bearer {}", token))
                .header("Content-Type", "application/octet-stream")
                .body(file_content.into_bytes())
                .send()
                .await
                .expect("Failed to send DVC PUT request");

            (file_path, dvc_put_response.status())
        });
    }

    // Collect all push results
    let mut successful_pushes = 0;

    while let Some(result) = push_tasks.join_next().await {
        let (file_path, status) = result.expect("Push task should complete successfully");
        assert_eq!(status, 200, "DVC PUT should succeed for {}", file_path);
        successful_pushes += 1;
    }

    assert_eq!(
        successful_pushes, file_count,
        "All concurrent DVC pushes should succeed"
    );
    println!(
        "Successfully completed {} concurrent DVC pushes for single user",
        successful_pushes
    );

    // Verify all files are present
    let dvc_list_response = server
        .client
        .get(format!("{}/api/dvc/files", server.base_url))
        .header("Authorization", format!("Bearer {}", token))
        .send()
        .await
        .expect("Failed to send DVC list request");

    assert_eq!(dvc_list_response.status(), 200);

    let dvc_list_body: Value = dvc_list_response
        .json()
        .await
        .expect("Failed to parse DVC list response");
    let files = dvc_list_body
        .as_array()
        .expect("DVC list should return an array");
    assert_eq!(
        files.len(),
        file_count,
        "All pushed files should be in the list"
    );

    println!("Verified all {} files are present in DVC list", file_count);
}

#[tokio::test]
async fn test_mixed_dvc_operations_concurrent() {
    let server = ConcurrentDvcTestServer::new().await;
    server
        .wait_for_server()
        .await
        .expect("Test server failed to start");

    // Register two users
    let token1 = server.register_and_login("user1", "password123").await;
    let token2 = server.register_and_login("user2", "password123").await;

    let mut mixed_tasks = JoinSet::new();

    // User 1: Push files
    for file_idx in 1..=5 {
        let file_path = format!("data/user1/push_{}.txt", file_idx);
        let file_content = format!("User 1 push file {}", file_idx);
        let server_clone = server.clone();
        let token = token1.clone();

        mixed_tasks.spawn(async move {
            let response = server_clone
                .client
                .put(format!(
                    "{}/api/dvc/remote/{}",
                    server_clone.base_url, file_path
                ))
                .header("Authorization", format!("Bearer {}", token))
                .header("Content-Type", "application/octet-stream")
                .body(file_content.into_bytes())
                .send()
                .await
                .expect("Failed to send DVC PUT request");

            ("push", file_path, response.status())
        });
    }

    // User 2: Push different files
    for file_idx in 1..=5 {
        let file_path = format!("data/user2/push_{}.txt", file_idx);
        let file_content = format!("User 2 push file {}", file_idx);
        let server_clone = server.clone();
        let token = token2.clone();

        mixed_tasks.spawn(async move {
            let response = server_clone
                .client
                .put(format!(
                    "{}/api/dvc/remote/{}",
                    server_clone.base_url, file_path
                ))
                .header("Authorization", format!("Bearer {}", token))
                .header("Content-Type", "application/octet-stream")
                .body(file_content.into_bytes())
                .send()
                .await
                .expect("Failed to send DVC PUT request");

            ("push", file_path, response.status())
        });
    }

    // Give some time for pushes to complete
    tokio::time::sleep(Duration::from_millis(500)).await;

    // User 1: List files
    for list_idx in 1..=3 {
        let server_clone = server.clone();
        let token = token1.clone();

        mixed_tasks.spawn(async move {
            let response = server_clone
                .client
                .get(format!("{}/api/dvc/files", server_clone.base_url))
                .header("Authorization", format!("Bearer {}", token))
                .send()
                .await
                .expect("Failed to send DVC list request");

            ("list", format!("list_{}", list_idx), response.status())
        });
    }

    // User 2: List files
    for list_idx in 1..=3 {
        let server_clone = server.clone();
        let token = token2.clone();

        mixed_tasks.spawn(async move {
            let response = server_clone
                .client
                .get(format!("{}/api/dvc/files", server_clone.base_url))
                .header("Authorization", format!("Bearer {}", token))
                .send()
                .await
                .expect("Failed to send DVC list request");

            ("list", format!("list_{}", list_idx), response.status())
        });
    }

    // Collect all results
    let mut push_count = 0;
    let mut list_count = 0;

    while let Some(result) = mixed_tasks.join_next().await {
        let (operation, identifier, status) =
            result.expect("Mixed task should complete successfully");
        assert_eq!(
            status, 200,
            "Mixed operation {} should succeed for {}",
            operation, identifier
        );

        match operation {
            "push" => push_count += 1,
            "list" => list_count += 1,
            _ => panic!("Unexpected operation: {}", operation),
        }
    }

    assert_eq!(push_count, 10, "All push operations should succeed");
    assert_eq!(list_count, 6, "All list operations should succeed");

    println!(
        "Successfully completed {} push and {} list operations concurrently",
        push_count, list_count
    );
}

// Implement Clone for the test server
impl Clone for ConcurrentDvcTestServer {
    fn clone(&self) -> Self {
        Self {
            port: self.port,
            temp_dir: TempDir::new().unwrap(), // Create a new temp dir for the clone
            base_url: self.base_url.clone(),
            client: self.client.clone(),
        }
    }
}
